FastAPI でRESTful Webサービスの実装 Part2
モデルクラス
APIに必要で重要な機能には、データ検証があります。Part1 ではリクエストデータの検証については不十分で、実際の運用には問題があります。
FastAPI では pydantic をサポートしていて、モデルクラスをPythonのタイプアノテーションで記述することで、リクエスト/レスポンスのデータについて検証を簡単に行うことができるようになっています。 pydantic でのデータの検証は簡単です。まず、モデルクラスを定義しておき、ルーティングしたビュー関数ではタイプアノテーションをします。
code: Python
from fastapi import FastAPI, HTTPException
from tasks import tasks
from pydantic import BaseModel
class Task(BaseModel):
title: str
description: str = None
done: bool = False
app = FastAPI()
# ...
@app.post("/todo/api/v1.0/tasks")
def create_task(request_data: Task):
tasks.append(request_data)
return {"data": request_data}
複数のモデルクラスを定義して使用することができます。
code: schema_test.py
from pydantic import ValidationError
from schemas import TaskSchema
try:
TaskSchema()
except ValidationError as e:
print(e.json())
検証エラー（ValidationError) が発生するとJSON形式で情報が渡されます。
code: bash
$ python schema_test.py
[
{
"loc": [
"title"
],
"msg": "field required",
"type": "value_error.missing"
}
]
loc：検証エラーが発生したフィールド名
msg：エラーメッセージ
type： エラーのタイプ
パラメタの検証
fastapi.Query および fastapi.Pathを使って引数の検証を強制することができます。
次のコードはよく似ていますが意味が異なります。
code: python
from fastapi import FastAPI
# ...
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(task_id: int=None):
# ...
code: python
from fastapi import FastAPI, Query
# ...
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(task_id: int=Query(None):
# ...
code: python
from fastapi import FastAPI
# ...
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(id: int):
# ...
code: python
from fastapi import FastAPI, Path
# ...
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(task_id: int=Path(...):
# ...
Query() はクエリパラメタを検証し、Path() はパスパラメタを検証します。
それぞれ最初の引数はデフォルト値として理解されます。
Query()にデフォルト値としてNone を設定したときは、デフォルトではなくパラメーターが省略可能として処理されます。デフォルトを持たず、パラメーターを必須にするために、Ellipsisもしくは３つのドット記号（...) を使用します。
Path() で処理するパスパラメタはURIの一部なので省略することができないため、ドットを記述する必要があることに留意してください。
つまり、Path() を使用すると関数の引数のデフォルト値を与えたことになります。
Path()を使用してパスパラメータを宣言して、Query()でデフォルト値も指定せずにクエリパラメータを宣言するときには注意が必要です。
Python では、関数でデフォルト値を持たない引数の前に、デフォルト値を持つ引数を定義することができません。
code: python
# SyntaxError: non-default argument follows default argument
def func(foo='python', bar):
pass
FastAPIでは、変数名、タイプ、デフォルトの宣言（Query、Pathなど）によってパラメーターを検出するため、順序は関係ありません。
まず、関数の最初のパラメーターとしてアスタリスク(*)を定義します。
Pythonはそのアスタリスク(*)を使用して何もしませんが、それ以降のすべてのパラメーターはキーワード引数として呼び出されるものと判断します。FastAPIもこれを利用しているため、Query() でデフォルト値を指定しなくてもうまく処理することができます。
code: Python
from fastapi import FastAPI, Path, Query
# ...
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(*, id: int=Path(...), task_id: int=Query(None):
# ...
パラメタのメタデータ
FastAPI はパラメタにメタデータを定義することができます。
title=STR：タイトルを文字列STRで与える
description=STR：説明を文字列STRで与える
alias=STR：別名を文字列STRで与える
deprecated=BOOL：BOOLが True であれば非推奨パラメタとして明示する
パラメタの別名
alias はURIで与えるクエリパラメタで Python で使用できないパラメタ名を指定できるようにするためのものです。
例えば、次のような場合です。
クエリパラメタ item-query はマイナス記号(-) が含まれているため、 Python では変数名として使用することができません。
http://127.0.0.1:8000/items/?item-query=foobaritems
次のように alias を定義することで、関数としては、変数名は item として定義しておき、URIではitem-query を使えるようになります。
code: Python
@app.get("/items/")
def read_items(item: str = Query(None, alias="item-query")):
# ...
非推奨のパラメタ
APIの開発をすすめていく中で、パラメタが仕様から除外され使われなくなることもあります。
ただし、現時点でAPIを利用しているユーザがいるわけなので、非推奨のパラメタとして明示しておくと便利です。こうしたときは、deprecated=True を与えておきます。
code: python
from fastapi import FastAPI
app = FastAPI()
def read_items():
@app.get("/elements/", tags="items", deprecated=True) def read_elements():
文字列の検証
パラメタが文字列型のとき、Path() および Query() は次の引数で文字列を検証することができます。
min_length=VAL：文字列の長さが VAL より短いときはエラー
max_length=VAL：文字列の長さが VAL より長いときはエラー
regex=RE：文字列が正規表現 RE にマッチしなければエラー
数値の検証
パラメタが整数型(int)や浮動小数点型(float)のとき、Path() および Query() は次の引数で数値範囲を検証することができます。
gt=VAL： VAL で与えた数値より大きくなければエラー (Greater Than)
ge=VAL：VAL で与えた数値以上でなければエラー (Greater Equal)
lt=VAL： VAL で与えた数値より小さくなければエラー (Less Than)
le=VAL：VAL で与えた数値以下でなければエラー (Less Equal)
code: Python
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(
id: int = Path(..., title="The ID of the task to get", gt=0)
):
"""This is GET REST API"""
task = [task for task in tasks if task'id' == id] if len(task) == 0:
raise HTTPException(status_code=404, detail="Not found")
else:
code: bash
{
"detail": [
{
"loc": [
"path",
"id"
],
"msg": "ensure this value is greater than 0",
"type": "value_error.number.not_gt",
"ctx": {
"limit_value": 0
}
}
]
}
code: python
from fastapi import FastAPI, HTTPException, Path
from tasks import tasks
from pydantic import BaseModel
class Task(BaseModel):
title: str
description: str = None
done: bool = False
app = FastAPI()
@app.get("/todo/api/v1.0/tasks/{id}")
def get_task(id):
"""This is GET REST API"""
try:
task_id = int(id)
except:
raise HTTPException(status_code=404, detail="Not found")
FastAPI は検証の方法については、pydantic を参考にして実装しているので、これらの検証で使用する引数は pydantic とほとんど同じになります。
リクエストデータの検証
リクエストボディーのデータについては、モデルクラスで検証することができます。
モデルクラスを複数定義して、それとは別のパラメタをもたせたいような場合を考えてみましょう。
例えば、リクエストデータが次のようなJSONで渡されるとします。
code: json
{
"item": {
"name": "Foo",
"description": "The pretender",
"price": 42.0,
"tax": 3.2
},
"user": {
"username": "dave",
"full_name": "Dave Grohl"
},
"importance": 5
}
item と user については、モデルクラスを定義すればよいのですが、importance は
そのままでは、クエリパラメタと認識されてしまうので、fastapi.BODY() を使用します。
code: python
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
class User(BaseModel):
username: str
full_name: str = None
@app.put("/items/{item_id}")
async def update_item(
*, item_id: int, item: Item, user: User, importance: int = Body(...)
):
モデルクラスのフィールドを検証
モデルクラスを定義するとき、pydantic.Field() を使うと、フィールドの値を検証することができます。
code: python
from fastapi import FastAPI, HTTPException
from tasks import tasks
from pydantic import BaseModel. Field
class Task(BaseModel):
title: str = Field(...,
title='The title of task', max_length=18)
description: str = Field(None,
title='The description of task', max_length=128)
done: bool = Field(False, title='The status of task')
app = FastAPI()
# ...
@app.post("/todo/api/v1.0/tasks")
def create_task(request_data: Task):
tasks.append(request_data)
return {"data": request_data}
環境変数の読み込み
pydantic は python-dotenv がインストールされていれば、pydantic.BaseSettingsを継承したクラスで設定されていない値は環境変数から読み込みます。
code: bash
$ pip install python-dotenv
$ echo "DATABASE_URI='sqlite:///todo.db' > .env
code: python
from pydantic import BaseSettings
class Settings(BaseSettings):
database_uri: str
class Config:
env_file = '.env'
case_sensitive = False # this is default
settings = Settings()
print(settings.database_uri)
APIをモジュールに分割
実用的なアプリケーションやRESTful Webサービスを開発するとき、コードが１つのファイルに収まることはほとんどありません。いくつかのモジュールに分割していくことになります。FastAPIではこうしたときに柔軟に対応できる機能を提供しています。
例えば、Webアプリケーションが次のようなディレクトリ構造で構成されているとします。
code: bash
.
├── app
│ ├── __init__.py
│ ├── main.py
│ └── routers
│ ├── __init__.py
│ ├── items.py
│ └── users.py
__init__.py が存在するディレクトリは、Python はモジュールとして読み込むときに対象にすることができます。
これまでは、FastAPI() で生成したインスタンオブジェクトで、ビュー関数をデコレートする形式でした。
code: python
from fastapi import FastAPI
from .user import router as user_router
app = FastAPI()
@app.get('/user')
def get_users():
# ...
FastAPI では APIRouter() で生成したインスタンスオブジェクトのinclude_router()メソッドでビュー関数を登録することができるので、APIをモジュールに分割しやすくなります。
code: python
from fastapi import APIRouter
from .user import router as user_router
router = APIRouter()
router.include_router(
user_router,
prefix='/user',
)
この場合、prefix=/user が定義されているため、 user.py では次のようにGETメソッドのルーティングを定義するだけでOKです。
code: python
from fastapi import APIRouter
router = APIRouter()
@router.get('/')
def user(
*,
user_id: int = Query(..., title="The ID of the user to get", gt=0),
response_model=User
):
my_user = get_user(user_id)
return my_user